package com.ruoyi.flink.util.write;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import com.ruoyi.flink.chapten5.Event;

/**
 * @Description: flink 向mysql中写入数据
 * @ClassName: MysqlWrite
 * @author syc
 * @date: 2022年5月30日 下午2:29:03
 */
public class MysqlWrite extends RichSinkFunction<Event>{
	
	private static final long serialVersionUID = 1L;

	private Connection connection = null;
	private PreparedStatement insertSmt = null;

	 @Override
     public void open(Configuration parameters) throws Exception {
		 super.open(parameters);
		 //Class.forName("com.mysql.cj.jdbc.Driver");   // 这个似乎可要可不要
		 String url = "jdbc:mysql://127.0.0.1:3306/mytest?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
         connection = DriverManager.getConnection(url,"root","19901110");
         insertSmt = connection.prepareStatement("INSERT INTO clicks (user, url) VALUES (?, ?)");
     }

     @Override
     public void invoke(Event value, Context context) throws Exception {
    	 //可以在这里做一些复杂的处理
    	 
         //直接执行更新语句
         insertSmt.setString(1,value.user);
         insertSmt.setString(2,value.url);
         insertSmt.execute();
     }
     @Override
     public void close() throws Exception {
    	 if (insertSmt != null) {
    		 insertSmt.close();
    	 }
    	 if (connection != null) {
    		 connection.close();
    	 }
     }
     
}
